package storm.online.bolt;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Iterator;
import java.util.Map;

import storm.online.common.LoginInfo;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class GroupOnlineBolt extends BaseRichBolt{

	private OutputCollector collector;
	private Map<String, Long> groupOnlineMap; //这个用来储存玩家的状态，0表示离线，1表示在线
	String filepath = "GroupOnlineNum.txt";
	
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		this.collector = collector;
	}
	
	private int writeRetFile(String path){
		try {
			File file = new File(path);
			
			if(file.exists()){
				if( !file.delete() ){
					return -1;
				}
			}
			
			if( !file.createNewFile() ){
				return -1;
			}
			
			String retString = "";
			
			Iterator iter = groupOnlineMap.entrySet().iterator();
			while(iter.hasNext()){
				 Map.Entry entry = (Map.Entry) iter.next();
				 String key = (String)entry.getKey();
				 Long onlineNum = (Long)entry.getValue();
				 
				 retString += key + "\t" + onlineNum + "\n";
			}
			
			BufferedWriter output = new BufferedWriter(new FileWriter(file));
			output.write(retString);
			return 0;
		} catch (IOException e) {
			e.printStackTrace();
			return -1;
		} catch (Exception e){
			e.printStackTrace();
			return -1;
		}
	}
	
	public void execute(Tuple tuple) {
		
		String gameId = tuple.getString(0);
		Long areaId = tuple.getLong(1);
		Long groupId = tuple.getLong(2);
		long numChange = tuple.getInteger(3);
		
		long onlineNum = 0;
		
		String key = gameId + "_" + areaId + "_" + groupId;
		if( !groupOnlineMap.containsKey(key) ){
			if( numChange > 0 ){
				onlineNum = numChange;
				groupOnlineMap.put(key, onlineNum);
			}
			else{
				onlineNum = 0;
				groupOnlineMap.put(key, onlineNum);
			}
		}
		else{
			onlineNum = groupOnlineMap.get(key);
			onlineNum += numChange;
			groupOnlineMap.put(key, numChange);
		}
		
		collector.emit( new Values(gameId, areaId, groupId, onlineNum));
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare( new Fields("gameId", "areaId", "groupId", "onlineNum"));
	}
	
    @Override
    public void cleanup() {
    	writeRetFile(filepath);
    }  

}
